package org.ccay.async.conduit;

import java.util.List;
import java.util.Properties;

import javax.inject.Inject;

import org.springframework.util.Assert;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;

import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.SerializationUtils;

import org.ccay.async.AsyncMessage;
import org.ccay.async.IAsyncFramework;
import org.ccay.async.config.AsyncConduitConfig;
import org.ccay.core.log.ILogger;
import org.ccay.core.log.CcayLoggerFactory;

/**
 * kafka消息通道
 * 
 * @since 2015-12-25
 */
public class KafkaConduit extends AbstractAsyncConduit implements InitializingBean {
	
	private final ILogger logger = CcayLoggerFactory.getLogger(KafkaConduit.class);
	
	@Inject
	private IAsyncFramework asyncFramework;
	
	private Producer<String, Object> producer;
	
	private KafkaProducerFactory producerFactory;
	
	private String topic;

	
	/**
	 * 初始化配置
	 */
	public void initialize() {
		Assert.notNull(producerFactory, "producer facotry must not be null");
		producer = producerFactory.createProducer();
	}

	public void handle(AsyncMessage message) {
		
		AsyncConduitConfig conduitConfig = asyncFramework.getConduitConfig(message);
		if (topic != null) {
			//序列化消息对象
			KeyedMessage<String, Object> data = new KeyedMessage<String, Object>(topic, SerializationUtils.serialize(message));
			producer.send(data);
		} else {
			logger.warn("not found messageType:" + message.getType());
		}
	}

	public void afterPropertiesSet() throws Exception {
		initialize();
	}

	public void setProducerFactory(KafkaProducerFactory producerFactory) {
		this.producerFactory = producerFactory;
	}

	public void setTopic(String topic) {
		this.topic = topic;
	}

}
